/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.postbar.handler.bar;

import cn.ac.iie.di.ban.ycl.idgen.IDGeneratorInterface;
import cn.ac.iie.di.ban.ycl.idgen.IDSHAGeneratorImpl;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import cn.ac.iie.di.postbar.entity.BarEntity;
import cn.ac.iie.di.postbar.handler.bar.BarHandler.SessionPool;
import static cn.ac.iie.di.postbar.server.PostLtStartUp.requestServerSize;
import static cn.ac.iie.di.postbar.server.PostLtStartUp.isTianYa;
import cn.ac.iie.di.postbar.util.Base64;
import cn.ac.iie.di.postbar.util.DeDuplication;
import cn.ac.iie.di.postbar.util.ParserJson;
import cn.ac.iie.di.postbar.util.TransformUtil;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import net.sf.json.JSONArray;
import org.apache.log4j.Logger;

/**
 *
 * @author Administrator
 */
public class BarThreadPoolTask implements Runnable {

    static private final Logger logger = Logger.getLogger(BarThreadPoolTask.class.getName());
    private final List<String> jsonList;
    private final int timeout;
    private final SessionPool pool;
    private long starttime = System.currentTimeMillis() / 1000;

    public BarThreadPoolTask(List<String> jsonList, SessionPool pool, int timeout) {
        this.jsonList = jsonList;
        this.pool = pool;
        this.timeout = timeout;
    }

    @Override
    public void run() {
        RESendSession session = null;

        //计数
        long count = 0;
        //请求去重服务字符串
        String requestString = "";
        //存放bar对象
        List<BarEntity> barEntityList = new ArrayList<BarEntity>();

        try {
            session = pool.borrowSession();
            for (String json : jsonList) {
                if (json != null && !json.isEmpty()) {
                    //解析json
                    List<Map<String, String>> dataMapList = new ArrayList<Map<String, String>>();
                    try {
                        dataMapList = ParserJson.getContentByRecord_list(json);
                        logger.debug("parser json data successfully.");
                    } catch (Exception e) {
                        logger.error("@@@ERROR_MESSAGE@@@:" + json);
                        logger.error("parser data error,cause " + e.getMessage(), e);
                        return;
                    }
                    logger.info("the json has " + dataMapList.size() + " data.");
                    for (int i = 0; i < dataMapList.size(); i++) {
                        BarEntity barEntity = new BarEntity();
                        Map<String, String> dataMap = new HashMap<String, String>();
                        dataMap = dataMapList.get(i);

                        barEntity.setG_channel(2);
                        barEntity.setG_spec("M-BBS08-AI");
                        if (isTianYa == 1) {
                            barEntity.setG_asp("tianya.cn");
                            barEntity.setG_adp("tianya.cn");
                        } else if (isTianYa == 2) {//人民网强国论坛
                            barEntity.setG_asp("people.com.cn");
                            barEntity.setG_adp("people.com.cn");
                        } else if (isTianYa == 3) {//爱卡
                            barEntity.setG_asp("xcar.com.cn");
                            barEntity.setG_adp("xcar.com.cn");
                        } else if (isTianYa == 4) {//铁血
                            barEntity.setG_asp("tiexue.net");
                            barEntity.setG_adp("tiexue.net");
                        } else if (isTianYa == 5) {//中华
                            barEntity.setG_asp("china.com");
                            barEntity.setG_adp("china.com");
                        } else if (isTianYa == 0) {
                            barEntity.setG_asp("kdnet.net");
                            barEntity.setG_adp("kdnet.net");
                        }

                        String b_name = dataMap.get("BARNAME");
                        if (b_name == null || b_name.isEmpty() || "null".equals(b_name)) {
                            barEntity.setB_name("");
                        } else {
                            try {
                                b_name = new String(Base64.decodeFast(b_name), "UTF-8");
                                barEntity.setB_name(b_name);
                            } catch (Exception e) {
                                logger.error("(b_name)Decode base64 failed, cause " + e.getMessage(), e);
                                logger.warn("@@b_name:" + dataMap.get("BARNAME"));
                                barEntity.setB_name("");
                            }
                        }

                        String g_ch_key = b_name;
                        barEntity.setG_ch_key(g_ch_key);

                        //ID
                        String id = "";
                        try {
                            IDGeneratorInterface idge = new IDSHAGeneratorImpl();
//                            if (isTianYa == 1) {
//                                id = idge.getId("2", "M-BBS08-AI", "tianya.cn", "tianya.cn", g_ch_key);
//                            } else {
//                                id = idge.getId("2", "M-BBS08-AI", "kdnet.net", "kdnet.net", g_ch_key);
//                            }
                            if (isTianYa == 1) {
                                id = idge.getId("2", "M-BBS08-AI", "tianya.cn", "tianya.cn", g_ch_key);
                            } else if (isTianYa == 2) {//人民网强国论坛
                                id = idge.getId("2", "M-BBS08-AI", "people.com.cn", "people.com.cn", g_ch_key);
                            } else if (isTianYa == 3) {//爱卡
                                id = idge.getId("2", "M-BBS08-AI", "xcar.com.cn", "xcar.com.cn", g_ch_key);
                            } else if (isTianYa == 4) {//铁血
                                id = idge.getId("2", "M-BBS08-AI", "tiexue.net", "tiexue.net", g_ch_key);
                            } else if (isTianYa == 5) {//中华
                                id = idge.getId("2", "M-BBS08-AI", "china.com", "china.com", g_ch_key);
                            } else {
                                id = idge.getId("2", "M-BBS08-AI", "kdnet.net", "kdnet.net", g_ch_key);
                            }
                            barEntity.setG_id(id);
                        } catch (Exception e) {
                            logger.error("(g_id)Get id error,cause " + e.getMessage(), e);
                            logger.warn("@@g_ch_key:" + g_ch_key);
                            continue;
                        }

                        barEntity.setB_insert_time(System.currentTimeMillis() / 1000);

                        String u_register_time = dataMap.get("BARRGSTTIME");
                        if (u_register_time == null || u_register_time.isEmpty() || "null".equals(u_register_time)) {
                            barEntity.setB_register_time(-1);
                        } else {
                            try {
                                long str = TransformUtil.timeToLong(u_register_time);
                                barEntity.setB_register_time(str);
                            } catch (Exception e) {
                                logger.error("(u_register_time)Type conversion error,cause " + e.getMessage(), e);
                                logger.warn("@@u_register_time:" + dataMap.get("BARRGSTTIME"));
                                barEntity.setB_register_time(-1);
                            }
                        }

                        String u_description = dataMap.get("BARINFO");
                        if (u_description == null || u_description.isEmpty() || "null".equals(u_description)) {
                            barEntity.setB_description("");
                        } else {
                            try {
                                u_description = new String(Base64.decodeFast(u_description), "UTF-8");
                                barEntity.setB_description(u_description);
                            } catch (Exception e) {
                                logger.error("(u_description)Decode base64 failed, cause " + e.getMessage(), e);
                                logger.warn("@@u_description:" + dataMap.get("BARINFO"));
                                barEntity.setB_description("");
                            }
                        }

                        String u_fans_cnt = dataMap.get("BARFCSNUM");
                        if (u_fans_cnt == null || u_fans_cnt.isEmpty() || "null".equals(u_fans_cnt)) {
                            barEntity.setB_fans_cnt(-1);
                        } else {
                            try {
                                long str = Long.parseLong(u_fans_cnt);
                                barEntity.setB_fans_cnt(str);
                            } catch (Exception e) {
                                logger.error("(u_fans_cnt)Type conversion error,cause " + e.getMessage(), e);
                                logger.warn("@@u_fans_cnt:" + dataMap.get("BARFCSNUM"));
                                barEntity.setB_fans_cnt(-1);
                            }
                        }

                        //百度贴吧用户填充BARPOSTNUM；天涯凯迪无对应原域名，填默认值-1
                        //已经修改了，因为论坛类消息有这个域名
                        String u_message_pub_cnt = dataMap.get("BARPOSTNUM");
                        if (u_message_pub_cnt == null || u_message_pub_cnt.isEmpty() || "null".equals(u_message_pub_cnt)) {
                            barEntity.setB_message_pub_cnt(-1);
                        } else {
                            try {
                                long str = Long.parseLong(u_message_pub_cnt);
                                barEntity.setB_message_pub_cnt(str);
                            } catch (Exception e) {
                                logger.error("(u_message_pub_cnt)Type conversion error,cause " + e.getMessage(), e);
                                logger.warn("@@u_message_pub_cnt:" + dataMap.get("BARPOSTNUM"));
                                barEntity.setB_message_pub_cnt(-1);
                            }
                        }

                        String u_snapshottime = dataMap.get("SNAPSHOTTIME");
                        if (u_snapshottime == null || u_snapshottime.isEmpty() || "null".equals(u_snapshottime)) {
                            barEntity.setB_snapshot_time(-1);
                        } else {
                            try {
                                long str = TransformUtil.timeToLong(u_snapshottime);
                                barEntity.setB_snapshot_time(str);
                            } catch (Exception e) {
                                logger.error("(u_snapshot_time)Type conversion error,cause " + e.getMessage(), e);
                                logger.warn("@@u_snapshot_time:" + dataMap.get("SNAPSHOTTIME"));
                                barEntity.setB_snapshot_time(-1);
                            }
                        }

                        String u_bar_owner = dataMap.get("BAROWNER");
                        String[] owner = new String[]{};
                        if (u_bar_owner == null || u_bar_owner.isEmpty() || u_bar_owner.equals("[]") || "null".equals(u_bar_owner)) {
                            barEntity.setB_owner(owner);
                        } else {
                            try {
                                JSONArray jsonArray = JSONArray.fromObject(u_bar_owner);
                                owner = new String[jsonArray.size()];
                                for (int k = 0; k < jsonArray.size(); k++) {
                                    owner[k] = jsonArray.getString(k);
                                }
                                barEntity.setB_owner(owner);
                            } catch (Exception e) {
                                logger.error("(u_bar_owner)Field segmentation failed, cause " + e.getMessage(), e);
                                logger.warn("@@u_bar_owner:" + u_bar_owner);
                                barEntity.setB_owner(new String[]{});
                            }
                        }

                        String u_bar_member = dataMap.get("BARMEMBER");
                        String[] member = new String[]{};
                        if (u_bar_member == null || u_bar_member.isEmpty() || u_bar_member.equals("[]") || "null".equals(u_bar_member)) {
                            barEntity.setB_member(member);
                        } else {
                            try {
                                JSONArray jsonArray = JSONArray.fromObject(u_bar_owner);
                                member = new String[jsonArray.size()];
                                for (int k = 0; k < jsonArray.size(); k++) {
                                    member[k] = jsonArray.getString(k);
                                }
                                barEntity.setB_member(member);
                            } catch (Exception e) {
                                logger.error("(u_bar_member)Field segmentation failed, cause " + e.getMessage(), e);
                                logger.warn("@@u_bar_member:" + u_bar_member);
                                barEntity.setB_member(new String[]{});
                            }
                        }

                        //百度贴吧信息填充BARMEMROLE；天涯凯迪无对应的原域名，填默认值
                        //直接按照下面的处理方式就可以，因为本身就没有这个原域名，所以必然为空，所以添加的值就是一个空的字符串数组
                        String u_bar_member_role = dataMap.get("BARMEMROLE");
                        String[] memrole = new String[]{};
                        if (u_bar_member_role == null || u_bar_member_role.isEmpty() || u_bar_member_role.equals("[]") || "null".equals(u_bar_member_role)) {
                            barEntity.setB_member_role(memrole);
                        } else {
                            try {
                                JSONArray jsonArray = JSONArray.fromObject(u_bar_owner);
                                memrole = new String[jsonArray.size()];
                                for (int k = 0; k < jsonArray.size(); k++) {
                                    memrole[k] = jsonArray.getString(k);
                                }
                                barEntity.setB_member_role(memrole);
                            } catch (Exception e) {
                                logger.error("(u_bar_member_role)Field segmentation failed, cause " + e.getMessage(), e);
                                logger.warn("@@u_bar_member_role:" + u_bar_member_role);
                                barEntity.setB_member_role(new String[]{});
                            }
                        }

                        String u_bar_dir = dataMap.get("BARDIR");
                        if (u_bar_dir == null || u_bar_dir.isEmpty() || "null".equals(u_bar_dir)) {
                            barEntity.setB_dir("");
                        } else {
                            try {
                                u_bar_dir = new String(Base64.decodeFast(u_bar_dir), "UTF-8");
                                barEntity.setB_dir(u_bar_dir);
                            } catch (Exception e) {
                                logger.error("(u_bar_dir)Decode base64 failed, cause " + e.getMessage(), e);
                                logger.warn("@@u_bar_dir:" + dataMap.get("BARDIR"));
                                barEntity.setB_dir("");
                            }
                        }

                        //4.12新添加的两个字段
                        //论坛类的接入规范没有这两个字段，全都填默认值，String的默认值是null，long型的默认值填-1
                        barEntity.setB_state("null");
                        barEntity.setB_prcsstime(-1);

                        count++;
                        requestString += id + "\n";
                        barEntityList.add(barEntity);

                        //请求去重服务并发送数据
                        if (count % requestServerSize == 0 && count != 0) {
                            sendMessageToMQ(barEntityList, session, requestString);
                            requestString = "";
                            barEntityList.clear();
                        }
                    }//一条json数据
                }
            }//所有json数据
            if (!barEntityList.isEmpty()) {
                sendMessageToMQ(barEntityList, session, requestString);
                count = 0;
                requestString = "";
                barEntityList.clear();
            }
            logger.info("A total " + count + " data of the bar.");
        } catch (Exception ex) {
            logger.error("parser data error, cause " + ex.getMessage(), ex);
        } finally {
            if (session != null) {
                pool.returnSession(session);
            }
        }
    }

    public void sendMessageToMQ(List<BarEntity> barEntityList, RESendSession session, String requestString) {
        //请求去重服务返回的信息数组
        String[] returnIds = new String[]{};
        long delcount = 0;
        List<BarEntity> realList = new ArrayList<BarEntity>();
        try {
            logger.debug("requestString=" + requestString);
            returnIds = DeDuplication.deduplication(requestString);
        } catch (Exception e) {
            logger.error("request to de-duplication failed, cause " + e.getMessage(), e);
            return;
        }

        //删除重复数据
        for (int s = 0; s < returnIds.length; s++) {
            if ("1".equals(returnIds[s])) {
                delcount++;
                logger.debug("remove one repeat data.");
            } else {
                realList.add(barEntityList.get(s));
            }
        }
        logger.info("remove datas of the bar: " + delcount);

        //循环发送数据
        try {
            for (int b = 0; b < realList.size(); b++) {
                BarEntity entity = realList.get(b);
                if (entity != null) {
                    session.setString("g_id", entity.getG_id());
                    session.setLong("g_channel", entity.getG_channel());
                    session.setString("g_spec", entity.getG_spec());
                    session.setString("g_ch_key", entity.getG_ch_key());
                    session.setString("g_asp", entity.getG_asp());
                    session.setString("g_adp", entity.getG_adp());
                    session.setString("b_name", entity.getB_name());
                    session.setLong("b_insert_time", entity.getB_insert_time());
                    session.setLong("b_register_time", entity.getB_register_time());
                    session.setString("b_description", entity.getB_description());
                    session.setLong("b_fans_cnt", entity.getB_fans_cnt());
                    session.setLong("b_message_pub_cnt", entity.getB_message_pub_cnt());
                    session.setLong("b_snapshot_time", entity.getB_snapshot_time());
                    session.setStrings("b_owner", entity.getB_owner());
                    session.setStrings("b_member", entity.getB_member());
                    session.setStrings("b_member_role", entity.getB_member_role());
                    session.setString("b_dir", entity.getB_dir());
                    //新添加的两个字段
                    session.setString("b_state", entity.getB_state());
                    session.setLong("b_prcsstime", entity.getB_prcsstime());

                    try {
                        session.add();
                    } catch (Exception e) {
                        sendMSG(session);
                        session.add();
                    }

                    if (System.currentTimeMillis() / 1000 - starttime > timeout) {
                        starttime = System.currentTimeMillis() / 1000;
                        sendMSG(session);
                        logger.info("timeout.");
                    }
                }
            }
            sendMSG(session);
            logger.info("send datas of bar: " + realList.size());
        } catch (Exception ex) {
            logger.error("setSession error, cause " + ex.getMessage(), ex);
        }
    }

    public void sendMSG(RESendSession session) {
        while (true) {
            try {
                logger.info("flush...");
                session.flush();
                break;
            } catch (Exception ex) {
                logger.info("what's up buddy ? how's it going ?" + ex, ex);
            }
        }
    }
}
